package com.runa.monitor.platinterface.netservice.equipment;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.runa.monitor.comm.dto.Archive;
import com.runa.monitor.comm.dto.Meter;
import com.runa.monitor.comm.dto.Node;
import com.runa.monitor.comm.dto.Valve;
import com.runa.monitor.comm.entity.CommNetEquipment;
import com.runa.monitor.platinterface.backwork.function.AbstractTaskFunction;
import com.runa.monitor.platinterface.backwork.function.TaskFunction;
import com.runa.monitor.platinterface.base.BaseContainer;
import com.runa.persistence.util.SpringContextUtil;
import com.runa.monitor.platinterface.extend.equ.EquInstructService;
import com.runa.monitor.platinterface.netservice.connect.commstatus.CommStatus;
import com.runa.monitor.platinterface.platService.RepositoryService;
import com.runa.monitor.platinterface.task.RequestTask;
import com.runa.persistence.CollectionUtil;
import com.runa.persistence.task.RequestPendingTask;
import com.runa.persistence.task.TaskTypeEnum;
import com.runa.persistence.util.LogUtil;
import com.runa.protocol.util.ByteUtil;
import com.runa.rccprotocol.dto.AffirmDeny;
import com.runa.rccprotocol.dto.ConcentratorRequestData;
import com.runa.rccprotocol.dto.ConcentratorResponseData;
import com.runa.rccprotocol.dto.equdata.EquData;
import com.runa.rccprotocol.dto.generic.OriginalConcenRequestData;
import com.runa.rccprotocol.dto.param.ExtendParam;
import com.runa.rccprotocol.netservice.dtu.DtuResponse;
import io.netty.channel.Channel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author zhuj
 */
@Slf4j
public abstract class NetEqu<S> {

    protected long heartbeatTimestamp = 0;

    protected long heartbeatTimeout = 10 * 60 * 1000;

    private static final ConcurrentMap<String, Object> channelSerial = new ConcurrentHashMap<>();

    public static Object getSerial(String channel) {
        return channelSerial.get(channel);
    }

    @Getter
    private Integer service;

    private final S serial;

    private int timeout;

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }
    private Channel channel;
    protected TaskFunction autoUpFunction;

    protected RepositoryService repositoryService;

    public NetEqu(S serial, Integer timeout) {
        this.serial = serial;
        this.timeout = timeout == null || timeout <= 0 ? 5000 : timeout;
        repositoryService = SpringContextUtil.getBean(RepositoryService.class);
    }

    Lock lock = new ReentrantLock();

    public Node currNode;

    int retrySign = 0;
    int maxRetry = 0;

    protected RequestTask requestTask;

    private final Queue<RequestTask> requestQueue = new PriorityBlockingQueue<>();

    public void sendRequest(RequestTask requestTask) {
        requestQueue.offer(requestTask);
    }

    public void sendRequest(ConcentratorRequestData request, TaskFunction function) {
        requestQueue.offer(new RequestTask(request, function));
    }

    public void sendRequest(ConcentratorRequestData request, TaskFunction function, TaskTypeEnum type) {
        requestQueue.offer(new RequestTask(request, function, new RequestPendingTask(type)));
    }

    public void sendRequest(ConcentratorRequestData request, TaskFunction function, RequestPendingTask task) {
        requestQueue.offer(new RequestTask(request, function, task));
    }

    protected long lastTime;

    protected volatile boolean sending;

    public void requestPost() {
        lock.lock();
        try {
            if (sending) {
                if (lastTime > 0 && System.currentTimeMillis() - lastTime > timeout) {
                    if (retrySign < maxRetry) {
                        send(this.requestTask);
                        retrySign++;
                        lastTime = System.currentTimeMillis();
                    } else {
                        log.info("集中器——{} 指令超时 ", serial);
                        if (null != requestTask) {
                            try {
                                requestTask.getTaskFunction().setStatus(requestTask.getRequestData().getRemain());
                                requestTask.getTaskFunction().doApply(CommStatus.TimeOut, null);
                            } catch (Exception e) {
                                log.error("集中器——{} 指令超时 报错: ", serial, e);
                            }
                        }
                        this.requestTask = null;
                        this.retrySign = 0;
                        this.sending = false;
                        LogUtil.info((String) serial, "指令超时");
                    }
                }
            } else {
                RequestTask requestTask = getRequestTask();
                if (requestTask != null) {
                    send(requestTask);
                }
            }
        }  catch (Exception e) {
            log.error("SendRequest Error: ", e);
        } finally {
            lock.unlock();
        }
    }

    public void setRequestTask(RequestTask requestTask) {
        this.requestTask = requestTask;
    }

    public RequestTask getRequestTask() {
        if (null != requestTask && requestTask.notUse()) {
            return requestTask;
        }
        return requestQueue.poll();
    }

    public void send(RequestTask requestTask) {
        try {
            requestTask.setUsed(true);
            if (online()) {
                log.info("集中器——{} 发送指令: {}", serial, ByteUtil.byteToHexString(requestTask.getRequestData().getData()));
                this.requestTask = requestTask;
                RequestPendingTask.addTask((String)serial, requestTask.getTask());
                writeOutput(requestTask.getRequestData());
                this.lastTime = System.currentTimeMillis();
                sending = requestTask.getRequestData().isHaveResult();
                LogUtil.info((String)serial, "{} sending: {} channelId: {} nodeIds: {}", getInfo(), sending, channel.id(), getNodeId());
                LogUtil.info((String)serial, requestTask.getRequestData().getData(), "发送");
                EquInstructService.start((String)serial, requestTask.getRequestData());
            } else {
                requestTask.getTaskFunction().doApply(CommStatus.OffLine, requestTask.getRequestData().getRemain());
            }
        } catch (Exception e) {
            log.error("NetEqu.send: ", e);
        }
    }

    protected void writeOutput(ConcentratorRequestData requestData) throws InterruptedException {
        channel.writeAndFlush(requestData).sync();
    }

    private String getInfo() {
        if (null == requestTask) {
            return null;
        }
        if (requestTask.getTaskFunction() instanceof AbstractTaskFunction) {
            AbstractTaskFunction taskFunction = (AbstractTaskFunction) requestTask.getTaskFunction();
            return taskFunction.getMessage() + " task: " + taskFunction.getTask();
        }
        return null;
    }

    private String getNodeId() {
        try {
            if (null == requestTask.getRequestData() || null == requestTask.getRequestData().getRemain()) {
                return null;
            }
            Object remain = requestTask.getRequestData().getRemain();
            if (remain instanceof String) {
                return (String)remain;
            } else if (remain instanceof Node) {
                Node node = (Node)remain;
                return node.getId() + ":" + node.getSerial();
            } else if (remain instanceof List) {
                List list = (List)remain;
                if (CollectionUtil.isEmpty(list)) {
                    return null;
                }
                List<String> nodes = new ArrayList<>();
                for (Object o : list) {
                    if (o instanceof Node) {
                        Node node = (Node)o;
                        nodes.add(node.getId() + ":" + node.getSerial());
                    }
                }
                if (!nodes.isEmpty()) {
                    return String.join(",", nodes);
                }
            }
            return JSON.toJSONString(remain);
        } catch (Exception e) {
            log.error("NetEqu.getNodeId: ", e);
        }
        return null;
    }

    public void receive(ConcentratorResponseData responseData) {
        heartbeatUpdate();
        BaseContainer.netEquReceiveFunc(() -> concentratorResponseReceive(responseData));
    }


    public void receiveDtuResponse(DtuResponse dtuResponse) {
    }

    public void doApply(TaskFunction taskFunction, CommStatus status, Object o) {
        log.info("{} {} {} ", serial, status, JSON.toJSONString(o));
        taskFunction.doApply(status, o);
    }

    public void concentratorResponseReceive(ConcentratorResponseData responseData) {
        try {
            lock.lock();
            log.info("receive进入 --- {} {} {}", serial(), responseData.isUpward(), null != requestTask ? JSONObject.toJSONString(requestTask.getRequestData()) : null);
            if (responseData.isUpward() && autoUpFunction != null) {
                doApply(autoUpFunction, CommStatus.OK, responseData.getData());
            } else if (requestTask != null) {
                if (requestTask.getRequestData().check(responseData)) {
                    log.info("{} 上行确认成功 {}", serial, JSONObject.toJSONString(responseData));
                    requestTask.getTaskFunction().setStatus(requestTask.getRequestData().getRemain());
                    try {
                        if (responseData instanceof AffirmDeny) {
                            doApply(requestTask.getTaskFunction(), CommStatus.OK, responseData);
                        } else {
                            doApply(requestTask.getTaskFunction(), CommStatus.OK, responseData.getData());
                        }
                    } catch (Exception e) {
                        log.error("发送业务层失败！", e);
                    }
                    if (responseData.isTailFrame()) {
                        mockUnlock();
                    }
                } else {
                    log.info("{} 上行确认失败 ", serial);
                }
            }
        } catch (Exception exception) {
            log.error("concentratorResponseReceive: ", exception);
        } finally {
            lock.unlock();
        }
    }

    protected void mockUnlock() {
        this.retrySign = 0;
        this.sending = false;
        this.requestTask = this.requestTask.getNext();
    }

    public S serial() {
        return serial;
    }

    public void register(Channel channel) {
        if (channel != null) {
            LogUtil.info((String)serial, "注册channel: {} {}", channel.id(), channel.remoteAddress());
            if (channel.isActive() && this.channel != channel) {
                if (this.channel != null && this.channel.isActive()) {
                    close();
                }
                this.channel = channel;
                channelSerial.put(channel.id().asLongText(), serial);
            }
        }
    }

    public boolean online() {
        return channel != null && channel.isActive();
    }

    public boolean close() {
        try {
            if (channel != null) {
                channel.close();
                heartbeatTimestamp = 0;
                LogUtil.info((String)serial, "关闭channel: {}", channel.id());
            }
        } catch (Exception e) {
            log.error("{} 关闭channel失败", serial);
            return false;
        }
        return true;
    }

    public void timeout(Integer timeout) {
        this.timeout = timeout == null || timeout <= 0 ? 5000 : timeout;
    }

    public int timeout() {
        return timeout;
    }

    public void sendMessage(TaskFunction function, Object message) {
        if (!online()) {
            function.doApply(CommStatus.OffLine, null);
        }
        if (message instanceof String) {
            sendRequest(new OriginalConcenRequestData((String) serial, (String) message), function);
        }
    }

    public CommNetEquipment commNetEquipment;

    public void init(CommNetEquipment commNetEquipment) {
        this.commNetEquipment = commNetEquipment;
    }


    public boolean notChange(CommNetEquipment commNetEquipment) {
        return this.commNetEquipment.getCommNetService().getProtocol().equals(commNetEquipment.getCommNetService().getProtocol())
                && this.commNetEquipment.getModel().equals(commNetEquipment.getModel());
    }

    public abstract void readNodeEqu(List<Node> nodes, TaskFunction function, EquData.EquDataType type);

    public abstract void setNodeEquArchive(List<Archive> archives, TaskFunction function);

    public abstract void clearNodeEquArchive(TaskFunction function);

    //    public abstract void valveOnOffControl(List<S> valveSerial, boolean ifLock, Integer opening, TaskFunction function);
    public abstract void valveOnOffControl(List<Node> nodeList, boolean ifLock, Integer opening, TaskFunction function);

    public abstract void readNodeRecord(List<Integer> indexes, TaskFunction function);

    //    public abstract void sentPublicUnitRecord(List<S> valveSerial, Integer model, Map param, TaskFunction function);
    public abstract void sentPublicUnitRecord(List<Node> nodeList, Integer model, Map param, TaskFunction function);

    //    public abstract void sentPublicUnitModel(List<S> valveSerial, boolean model, TaskFunction function);
    public abstract void sentPublicUnitModel(List<Node> nodeList, boolean model, TaskFunction function);

    public abstract void sentExtendParam(ExtendParam param, TaskFunction function);

    //    public abstract void readPublicUnitRecord(S valveSerial, Integer model, TaskFunction function);
    public abstract void readPublicUnitRecord(Node valve, Integer model, TaskFunction function);

    //    public abstract void readPublicUnitAndMeter(S valveSerial, S meterSerial, TaskFunction function);
    public abstract void readPublicUnitAndMeter(Node valve, Node meter, TaskFunction function);

    public void heartbeatUpdate() {
        this.heartbeatTimestamp = System.currentTimeMillis();
    }

    public boolean heartStop() {
        return heartbeatTimestamp != 0 && System.currentTimeMillis() - heartbeatTimestamp > heartbeatTimeout;
    }

    public Node findMeter(String serial, String netId) {
        return Meter.getMeter(repositoryService.getCommMeterByNodeSerialForNetEqu(serial, netId));
    }

    public Node findValve(String serial, String netId) {
        return Valve.getValve(repositoryService.getCommValveByNodeSerialForNetEqu(serial, netId));
    }


    public NetEqu<S> setService(Integer service) {
        this.service = service;
        return this;
    }
}
